package org.easy.file.handler;

import cn.hutool.core.thread.NamedThreadFactory;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.Queue;
import java.util.concurrent.*;

/**
 * @Description: 文件队列
 * @Author: wxl
 * @CreateTime: 2022/12/29 9:12
 */
@Slf4j
public class FileQueue{

    private final static int CORE_POOL_SIZE = 2;

    private final static int MAX_POOL_SIZE  = 10;

    private final static int KEEP_ALIVE_TIME = 0;

    private final static int WORK_QUEUE_SIZE = 50;

    private final Map<String, FileParse> fileCacheMap = new ConcurrentHashMap<>();

    private final Queue<FileParse> fileQueue = new LinkedBlockingQueue<>();

    private final RejectedExecutionHandler handler = (r, executor) -> {
        if (r instanceof FileThread) {
            FileThread fileThread = (FileThread) r;
            fileQueue.offer(fileThread.getFileParse());
        }
    };

    private final ThreadPoolExecutor queueThreadPool = new ThreadPoolExecutor(CORE_POOL_SIZE,MAX_POOL_SIZE,KEEP_ALIVE_TIME,TimeUnit.SECONDS,new ArrayBlockingQueue<>(WORK_QUEUE_SIZE),this.handler);

    public FileQueue () {
        NamedThreadFactory namedThreadFactory = new NamedThreadFactory("execute-pool-file-",true);
        this.queueThreadPool.setThreadFactory(namedThreadFactory);
    }

    public void addFileParse(FileParse fileParse){
        // TODO 文件校验。对于非法的文件有，则告知并删除该文件
        log.info("添加任务:{}到队列中.....",fileParse.getFileId());
        fileCacheMap.putIfAbsent(fileParse.getFileId(), fileParse);
        FileThread fileThread = new FileThread(fileParse);
        queueThreadPool.execute(fileThread);
    }

    final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);

    final ScheduledFuture<?> scheduledFuture = scheduler.scheduleAtFixedRate(() -> {
        // 判断缓冲队列是否存在记录
        if (!fileQueue.isEmpty()) {
            // 当前线程的队列容量
            if (queueThreadPool.getQueue().size() < WORK_QUEUE_SIZE) {
                FileParse fileParse = fileQueue.poll();
                FileThread fileThread = new FileThread(fileParse);
                queueThreadPool.execute(fileThread);
            }
        }
    },0,1,TimeUnit.SECONDS);

    public Queue<FileParse> getFileQueue () {
        return fileQueue;
    }

    public void shutdown () {
        scheduler.shutdown();
        queueThreadPool.shutdown();
    }

    public static FileQueue create () {
        return new FileQueue();
    }

}
